package com.chtwm.redis.sync;

import com.alibaba.fastjson.JSONObject;
import com.chtwm.redis.sync.config.RedisConfig;
import com.chtwm.redis.sync.config.SyncConfig;
import com.chtwm.redis.sync.service.MyRedisSocketService;
import com.moilioncircle.redis.replicator.RedisReplicator;
import com.moilioncircle.redis.replicator.Replicator;
import com.moilioncircle.redis.replicator.event.PostRdbSyncEvent;
import com.moilioncircle.redis.replicator.rdb.datatype.KeyStringValueString;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.net.URISyntaxException;
import java.util.Properties;

/**
 * redis同步方法
 */
@Slf4j
public class RedisSyncMain {

    public static Marker fatal = MarkerFactory.getMarker("FATAL");

    /**
     * 入口
     *
     * @param args
     */
    public static void main(String[] args) throws Exception {
        RedisSyncMain redisSyncMain = new RedisSyncMain();
        SyncConfig syncConfig = redisSyncMain.refreshConfig();
        MDC.put("methodId", "RedisSync");
        log.info("读取到配置：{}", JSONObject.toJSONString(syncConfig));
//        redisSyncMain.sync(syncConfig);
        MyRedisSocketService rss = new MyRedisSocketService(syncConfig);
//        rss.syncAll();

        log.error("error-error");
        log.error(fatal, "fatal-MARKER");

        MDC.put("FATAL", "FATAL");
        log.error("fatal-MDC");
        MDC.remove("FATAL");
        
        log.info("info");

    }


    /**
     * 全量keys
     *
     * @param syncConfig
     * @throws IOException
     * @throws URISyntaxException
     */
    public void sync(SyncConfig syncConfig) throws IOException, URISyntaxException {
        Replicator replicator = new RedisReplicator(syncConfig.getSource().getHost());
        //设置offset
        replicator.getConfiguration().setReplOffset(syncConfig.getSyncOffset());

        replicator.addEventListener((repl, event) -> {
            log.info("offset-{}", repl.getConfiguration().getReplOffset());
            //全量数据
            log.info("监听到事件-{}", event.toString());
            if (event instanceof KeyStringValueString) {

                KeyStringValueString kv = (KeyStringValueString) event;
                log.info("监听到key-{},val-{}", new String(kv.getKey()), new String(kv.getValue()));

                syncConfig.setSyncOffset(repl.getConfiguration().getReplOffset());

            } else if (event instanceof PostRdbSyncEvent) {
                //AOF文件
                repl.addRawByteListener((byte... bytes) -> {
                    log.info("rawBytes-{}", new String(bytes));
                });

                syncConfig.setSyncOffset(repl.getConfiguration().getReplOffset());
            }
        });

        log.info("启动监听！");
        replicator.open();
    }


    /**
     * 读取配置文件
     *
     * @return
     */
    private SyncConfig refreshConfig() throws Exception {
        Properties properties = readProperties();
        SyncConfig syncConfig = new SyncConfig();
        RedisConfig sourceConfig = new RedisConfig();
        sourceConfig.setHost(properties.getProperty("redis.source.host"));
        RedisConfig destinationConfig = new RedisConfig();
        destinationConfig.setHost(properties.getProperty("redis.destination.host"));

        syncConfig.setSource(sourceConfig);
        syncConfig.setDestination(destinationConfig);
        return syncConfig;
    }

    /**
     * 读取Properties配置文件
     *
     * @return
     */
    private Properties readProperties() throws Exception {
        Properties properties = new Properties();
        InputStream inputStream = this.getClass().getClassLoader()
                .getResourceAsStream("config.properties");
        properties.load(inputStream);
        return properties;
    }
}
